package com.atguigu.day04;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import com.atguigu.utils.UrlViewCountPerWindow;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

// 每个url在每个5秒钟的滚动窗口中被浏览的次数
public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<ClickEvent, UrlViewCountPerWindow, String, TimeWindow>() {
                    // 时间到达窗口结束时间-1毫秒时，触发process的执行
                    @Override
                    public void process(String key, Context context, Iterable<ClickEvent> elements, Collector<UrlViewCountPerWindow> out) throws Exception {
                        // 迭代器中包含了窗口中的所有元素
                        // 为了计算url的访问量，我们将窗口中所有数据都保存了下来
                        long count  = 0L;
                        for (ClickEvent e : elements) count++;
                        out.collect(new UrlViewCountPerWindow(
                                key,
                                count,
                                context.window().getStart(),
                                context.window().getEnd()
                        ));
                    }
                })
                .print();

        env.execute();
    }
}
